package com.zallds.window.trigger;

import com.zallds.utils.Utils;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ReducingState;
import org.apache.flink.api.common.typeutils.base.IntSerializer;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.api.common.state.ReducingStateDescriptor;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

/**
 * Flink window api中自定义触发器演示
 * 为了方便游客，故宫出口处依次停放着摆渡车，用来将游客送到故宫出口处。
 *  摆渡车出发有下列两种情形（满足下述任意一种情形，摆渡车就触发）：
 *   情形1：旅客坐满了 （约定一辆摆渡车最多能够载4人，真实的业务场景是：10人）
 *  * 情形2：停留的时间到了 （每辆摆渡车最多停留3秒钟，真实的业务场景是：一刻钟）
 */
public class SelfTriggerDemo {

    //摆渡车停留的时长，单位：秒
    static int MAX_WAIT_TIME = 3;

    //摆渡车最多能搭乘的旅客人数，单位：个
    static int MAX_PERSON_CNT = 4;

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.socketTextStream("192.168.62.37", 8889)
                .filter( keyword -> StringUtils.isNotEmpty(keyword))
                .flatMap(new Utils.Splitter())
                .keyBy(value -> value.f0)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(MAX_WAIT_TIME)))
                .trigger(new MyTrigger(MAX_PERSON_CNT))
                .maxBy(1)  // sum()
                .print();


        env.execute();
    }


    private static class MyTrigger extends Trigger<Tuple2<String, Integer>, TimeWindow> {

        int MAX_PERSON_CNT;

        private MyTrigger(int maxPersonCnt) {
            this.MAX_PERSON_CNT = maxPersonCnt;
        }


        /**
         * 用于储存窗口当前数据量的状态对象
         */
        private ReducingStateDescriptor<Integer> countState = new ReducingStateDescriptor<Integer>("counter", new Sum(), IntSerializer.INSTANCE);


        /**
         * @param element   抵达的元素，此处指的是：当前出故宫想要乘坐摆渡车的旅客
         * @param timestamp 元素抵达的时间戳
         * @param window    元素被添加到的窗口
         * @param ctx       →触发器的上下文信息
         * @return
         * @throws Exception
         */
        @Override
        public TriggerResult onElement(Tuple2<String, Integer> element, long timestamp, TimeWindow window, TriggerContext ctx) throws Exception {

            //注册处理时间
            ctx.registerProcessingTimeTimer(window.maxTimestamp());

            //a)获得ReducingState的实例
            ReducingState<Integer> state = ctx.getPartitionedState(countState);
            state.add(1);

            // b)获得状态中的值
            Integer value = state.get();
            if (value >= MAX_PERSON_CNT) {
                System.out.println("本质：计数触发；业务场景→ 摆渡车坐满旅客了，要发车了哦！...");

                state.clear();
                return TriggerResult.FIRE_AND_PURGE;
            } else {
                return TriggerResult.CONTINUE;
            }
        }

        /**
         * 计时触发，时间是：处理事件
         * @param time
         * @param window
         * @param ctx
         * @return
         * @throws Exception
         */
        @Override
        public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
            System.out.println("本质：计时触发；业务场景→ 摆渡车等待了指定的时间，即使旅客没有坐满，也要发车了哦！...");
            return TriggerResult.FIRE_AND_PURGE;
        }

        /**
         *
         * @param time
         * @param window
         * @param ctx
         * @return
         * @throws Exception
         */
        @Override
        public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
            return TriggerResult.CONTINUE;
        }

        /**
         *  Clears any state that the trigger might still hold for the given window.
         *    清空窗口中的元素
         * @param window
         * @param ctx
         * @throws Exception
         */
        @Override
        public void clear(TimeWindow window, TriggerContext ctx) throws Exception {
            ctx.getPartitionedState(countState).clear();
            ctx.deleteProcessingTimeTimer(window.maxTimestamp());
        }



        /**
         * 该触发器是否支持合并
         * @return
         */
        @Override
        public boolean canMerge() {
            return true;
        }

        /**
         * 多个TaskManager进程中的slot所维护的线程实现合并
         * @param window
         * @param ctx
         * @throws Exception
         */
        @Override
        public void onMerge(TimeWindow window, OnMergeContext ctx) throws Exception {
            ctx.mergePartitionedState(countState);
        }
    }

    /**
     * 计数方法
     */
    private static class Sum implements ReduceFunction<Integer> {

        @Override
        public Integer reduce(Integer value1, Integer value2) throws Exception {
            return value1 + value2;
        }
    }

}
